到第 13 篇為止，Relix 的 pipeline 是同步的，handler 和 middleware 都是 RelixCall.() -> RelixResponse，只要你開始做資料庫存取或呼叫外部 API，就會希望 handler 能用 suspend，把等待 I/O 的時間交給協程調度，而不是阻塞整條 thread

這篇要做的事，把 pipeline 的型別全面升級成 suspend，確保既有測試全部還是沒有問題的

為什麼要升級成 suspend ?

同步 handler 會讓等待 I/O 的期間也占著 executor thread，JDK HttpServer 的實際並行方式由設定的 Executor 決定，不論使用哪種 executor，框架內部支援 suspend 後，middleware 才能自然呼叫協程的 API

系列採取的策略是「先同步再非同步」，先把概念講清楚，再把型別升級，這樣學起來比較不會被協程細節淹沒

加 coroutines dependency

動程式碼之前，先把相依性補上，這也是系列第一次動到 module.yaml 的 dependencies

第 02 篇建出來的專案，module.yaml 從頭到尾只有一行

product: jvm/app

typealias 加 suspend 這件事本身不需要任何函式庫，suspend 是 Kotlin 的語言關鍵字，編譯器產生的 Continuation 也在 kotlin-stdlib 裡，但等一下 TestKit 和 JDK adapter 都要用 runBlocking 當 bridge，而 runBlocking 來自 kotlinx-coroutines-core，沒有它的話 import kotlinx.coroutines.runBlocking 那行會直接紅字

product: jvm/app

dependencies:
  - org.jetbrains.kotlinx:kotlinx-coroutines-core:1.11.0

這裡用的是當下的最新穩定版 1.11.0，你也可以到 MVN Repository 挑一個與自己 Kotlin Toolchain 相容的版本

加完先跑一次 ./kotlin build，確認相依性抓得到，再往下改型別

更新 typealias 型別

把原本在 Pipeline.kt 和 RelixHandler.kt 同步的 typealias 全部加上 suspend

// 同步版(第 12-13 篇)
typealias RelixHandler = RelixCall.() -> RelixResponse
typealias Next = RelixCall.() -> RelixResponse
typealias RelixMiddleware = RelixCall.(next: Next) -> RelixResponse

// suspend 版(本篇升級)
typealias RelixHandler = suspend RelixCall.() -> RelixResponse
typealias Next = suspend RelixCall.() -> RelixResponse
typealias RelixMiddleware = suspend RelixCall.(next: Next) -> RelixResponse

每行只多了一個 suspend，型別的結構完全沒有變，差別只在這些 lambda 現在可以呼叫 suspend function 了

buildPipeline 不用動

在 Pipeline.kt 裡面的 buildPipeline，一個字都不用改

fun buildPipeline(handler: Next, middlewares: List<RelixMiddleware>): Next {
    return middlewares.foldRight(handler) { middleware, next ->
        { middleware(next) }
    }
}

因為 Next 和 RelixMiddleware 的 typealias 已經改成 suspend，foldRight 裡面的 lambda 自動就是 suspend lambda。你也不需要在 buildPipeline 的函式簽名上加 suspend，因為它只是在組合函式，不是在執行它

這是 Kotlin 型別系統的好處，typealias 改了，依賴它的程式碼自動跟上

更新 RelixApplication.handle()

handle() 要變成 suspend function

class RelixApplication {

    suspend fun handle(call: RelixCall): RelixResponse {
        val terminal: Next = when (
            val result = router.match(call.request.method, call.request.path)
        ) {
            is MatchResult.Matched -> {
                call.pathParams = result.pathParams
                call.matchedRoute = result.route
                result.route.handler
            }
            is MatchResult.NotFound -> { { notFound() } }
            is MatchResult.MethodNotAllowed -> {
                { methodNotAllowed(result.allowedMethods) }
            }
        }

        val pipeline = buildPipeline(terminal, middlewares)
        val response = pipeline(call)

        return if (call.request.method == "HEAD") {
            response.copy(body = ByteArray(0))
        } else {
            response
        }
    }
}

唯一的差異是 fun handle 變成 suspend fun handle

如果你只改了 typealias，忘了改這裡，val response = pipeline(call) 這行會編譯失敗

Suspend function 'suspend fun invoke(p1: RelixCall): RelixResponse' can only be called from a coroutine or another suspend function.

原因是 pipeline 的型別是 Next，而 Next 已經加上 suspend 了，所以 pipeline(call) 變成一個 suspend 呼叫，它需要一個 suspend 的呼叫端，解法就是把 handle() 加上 suspend，函式裡面的邏輯一行都不用動

TestKit 升級，runBlocking bridge

handle() 變成 suspend 之後，TestKit 的 handleRequest() 不能直接呼叫它了，它會編譯錯誤，最簡單的 bridge 是用 runBlocking

import kotlinx.coroutines.runBlocking

class RelixTestKit(private val application: RelixApplication) {

    fun handleRequest(
        method: String = "GET",
        path: String = "/",
        headers: Map<String, List<String>> = emptyMap(),
        queryParameters: Map<String, List<String>> = emptyMap(),
        body: ByteArray = ByteArray(0),
    ): RelixResponse = runBlocking {
        val request = RelixRequest(
                   method = method,
                   path = path,
                   headers = headers,
                   queryParameters = queryParameters,
                   body = body,
        )
        val call = RelixCall(application, request)
        application.handle(call)
    }
}

handleRequest() 的簽名沒變，還是回傳 RelixResponse (不是 suspend)，內部用 runBlocking 把 suspend handle 跑完，這代表走 TestKit 的既有測試，呼叫方式完全不用改

testApplication helper 也不需要改，因為它回傳的是 RelixTestKit，而 RelixTestKit 內部處理了 runBlocking

JDK HttpServer Adapter 的 bridge

跟 TestKit 一樣，JDK adapter 也需要 runBlocking

class JdkHttpServerAdapter(private val application: RelixApplication) {
    private lateinit var server: HttpServer
    var port: Int = 0
        private set

    fun start(port: Int = 8080) {
        try {
            server = HttpServer.create(InetSocketAddress(port), 0)
        } catch (e: java.net.BindException) {
            throw IllegalStateException(
                "Port $port is already in use. Try a different port or use port 0 for auto-assign.",
                e
            )
        }

        server.createContext("/") { exchange ->
            try {
                val request = exchange.toRelixRequest()
                val call = RelixCall(application, request)
                val response = runBlocking { application.handle(call) }
                exchange.writeResponse(response)
            } catch (e: Exception) {
                val body = "Internal Server Error".toByteArray()
                exchange.sendResponseHeaders(500, body.size.toLong())
                exchange.responseBody.use { it.write(body) }
            }
        }

        server.start()
        this.port = server.address.port
    }

    fun stop() {
        server.stop(0)
    }
}

關鍵是 runBlocking { application.handle(call) }，JDK HttpServer 的 handler 是同步 callback，不在 coroutine context 裡，因此需要一個 bridge，這個版本會阻塞目前的 executor thread，suspend 本身不會把 JDK adapter 變成 non-blocking server

suspend 編譯後到底長什麼樣

只加一個關鍵字 pipeline 就升級了，看起來很簡單，但 compiler 背後做了不少事，Kotlin 會把 suspend function 編譯成 CPS (continuation passing style)，函式簽名多一個隱藏的 Continuation 參數，函式內部也會被拆成狀態機，每個 suspend point 對應一個 state，它不是讓 thread 神奇消失，而是把「等 I/O」改寫成「先保存目前狀態，之後再接著跑」

// 你寫的
suspend fun handle(call: RelixCall): RelixResponse {
    val response = pipeline(call)        // suspend point
    return response
}

// compiler 在 bytecode 層面做的事(概念，不是真的程式碼)
fun handle(call: RelixCall, $cont: Continuation<RelixResponse>): Any {
    when ($cont.label) {
        0 -> {
            $cont.label = 1
            val result = pipeline(call, $cont)            // 可能立即回值，也可能回 COROUTINE_SUSPENDED
            if (result == COROUTINE_SUSPENDED) return result
            return result
        }
        1 -> return $cont.result                          // 被 resume 時從這裡接著跑
    }
}

runBlocking 的職責是在目前 thread 上跑完一段 coroutine，每當 continuation resume 時，它會把流程接回去，直到整條 chain 結束，對非 coroutine 的呼叫端來說，像 JDK HttpServer 的 handler thread、JUnit 的同步測試，這是一座必要的橋，它不是長期架構的最佳解，只是把同步入口接到 suspend pipeline 的務實做法

如果你想再深入，Roman Elizarov 的 KotlinConf talk「Deep Dive into Coroutines on JVM」把整個編譯流程用組譯碼一行一行示範，系列裡不深入這塊，但知道「suspend 不是 thread」這件事，足以讓你寫出正確的 middleware

• KotlinConf 2017 - Deep Dive into Coroutines on JVM by Roman Elizarov

先修好編譯失敗的既有測試

在跑測試之前，有兩個測試檔會先卡住，第 12 篇的 MiddlewareTest 和第 13 篇的 PipelineTest，它們的共通點是都沒有經過 TestKit，而是自己直接呼叫 middleware 或 pipeline

PipelineTest 是自己組 pipeline、自己呼叫

val pipeline = buildPipeline(handler, listOf(middlewareA, middlewareB))
val call = RelixCall(RelixApplication(), RelixRequest("GET", "/", emptyMap(), emptyMap(), ByteArray(0)))
pipeline(call)    // 編譯失敗

MiddlewareTest 則是直接把 middleware 當函式呼叫，自己傳一個 lambda 扮演 next

val response = requestIdMiddleware(call) {    // 編譯失敗
    seenInsideHandler = context.get<String>("traceId")
    ok("done")
}

兩個失敗的原因是同一個，pipeline 的型別是 Next、requestIdMiddleware 的型別是 RelixMiddleware，兩個 typealias 現在都帶著 suspend，而 @Test fun 是普通函式，錯誤訊息跟前面 handle() 那次一模一樣

修法也一樣，把呼叫包進 runBlocking

import kotlinx.coroutines.runBlocking

// PipelineTest
val response = runBlocking { pipeline(call) }

// MiddlewareTest
val response = runBlocking {
    requestIdMiddleware(call) {
        seenInsideHandler = context.get<String>("traceId")
        ok("done")
    }
}

MiddlewareTest 有四個測試，PipelineTest 有五個，每一個都有自己的呼叫點，要各包一次，但只有呼叫那一行要動，其它的寫法都不用改

middleware 和 handler 的定義也都不用動，不管是 val middlewareA: RelixMiddleware = { next -> ... } 還是第 12 篇那三個示範 middleware，這種沒有呼叫 suspend function 的普通 lambda 會被隱式轉成 suspend lambda，裡面的 next() 也因為身處 suspend lambda 而合法

直接碰 middleware、pipeline 或 handle() 的測試要補 runBlocking，走 RelixTestKit.handleRequest() 的測試一行都不用改，因為 runBlocking 已經在 TestKit 內部了

確認既有測試沒壞，再補新測試

編譯過了之後，先跑一次完整的測試套件，前面累積下來的測試就是這次升級最好的安全網，這次改的是型別不是行為，行為沒變，它們就應該要全部通過

確認舊的沒壞之後，再補三個專門給 suspend 版的測試，把洋蔥模型的 before/after 順序、短路、routing 各驗一次，確認它們在 suspend 之後還是原本的樣子

import kotlin.test.Test
import kotlin.test.assertEquals

class SuspendPipelineTest {

    @Test
    fun `suspend pipeline maintains before-after order`() {
        val trace = mutableListOf<String>()

        val app = RelixApplication()
        app.use { next ->
            trace += "A-before"
            val response = next()
            trace += "A-after"
            response
        }
        app.use { next ->
            trace += "B-before"
            val response = next()
            trace += "B-after"
            response
        }
        app.routing {
            get("/test") {
                trace += "handler"
                ok("done")
            }
        }

        val testKit = RelixTestKit(app)
        val response = testKit.handleRequest("GET", "/test")

        assertEquals(200, response.statusCode)
        assertEquals(
            listOf("A-before", "B-before", "handler", "B-after", "A-after"),
            trace,
        )
    }

    @Test
    fun `suspend middleware can short-circuit`() {
        val app = RelixApplication()
        app.use { _ ->
            RelixResponse(401, emptyMap(), "Unauthorized".toByteArray())
        }
        app.routing {
            get("/secret") { ok("secret data") }
        }

        val testKit = RelixTestKit(app)
        val response = testKit.handleRequest("GET", "/secret")

        assertEquals(401, response.statusCode)
    }

    @Test
    fun `existing routing tests still pass after suspend upgrade`() {
        val app = RelixApplication()
        app.routing {
            get("/hello") { ok("Hello!") }
            get("/users/{id}") { ok("User: ${pathParam("id")}") }
        }

        val testKit = RelixTestKit(app)

        assertEquals(200, testKit.handleRequest("GET", "/hello").statusCode)
        assertEquals("User: 42", testKit.handleRequest("GET", "/users/42").bodyAsText())
        assertEquals(404, testKit.handleRequest("GET", "/missing").statusCode)
    }
}

遷移清單

做這種橫跨多個檔案的型別升級，列一張清單會比較不容易漏，順序就是實際動手的順序

module.yaml 要先加，runBlocking 沒有相依性就用不了

中間三個 src/ 檔案是型別升級的本體，接下來四個 bridge 現場其實是同一件事，凡是從非 suspend 的地方呼叫 suspend 的東西，都要有一座橋

RelixApplicationTest.kt 不在清單裡，因為它走的是 TestKit，runBlocking 已經在 handleRequest() 內部了

常見陷阱與設計取捨

runBlocking 不是不好嗎 ? 為什麼到處用 ?

runBlocking 會阻塞目前的 thread，因此不能放在 Netty 之類的 event loop 上，這裡只把它當成 JDK 同步 callback 與 suspend pipeline 之間的橋接，實際吞吐量仍受 executor 設定、阻塞式工作與 thread 數量影響，不能直接推論成與同步版完全相同

如果之後換成 Netty 或 CIO 引擎，就要把 runBlocking 換成 coroutine-native 的處理方式，但那是引擎層的事，framework 內部的 suspend pipeline 不用改

handler 裡不寫 suspend 的話會怎樣 ?

完全沒問題，普通的 lambda (沒有呼叫任何 suspend function) 可以隱式轉成 suspend lambda，所以之前寫的所有 handler { ok("Hello!") } 不用改，compiler 會自動處理

為什麼先同步再 suspend，不直接從 suspend 開始 ?

兩個原因，第一，如果同時學 pipeline 概念和 coroutine，注意力會分散，先用同步版把洋蔥模型、fold 串鏈、短路行為都搞懂，再加上 suspend，一次只改一個變數，第二，同步版的測試是 suspend 版的安全網，如果 suspend 升級出了問題，你有一整套同步版的測試可以對照

小結

升級 suspend 改的東西不多，三個 typealias 加 suspend、handle() 加 suspend、TestKit 和 Adapter 用 runBlocking bridge。buildPipeline 的組合邏輯完全不用動。走 TestKit 的既有測試不用改寫法，只有直接呼叫 middleware 或 pipeline 的 MiddlewareTest 和 PipelineTest 要補 runBlocking。先同步再 suspend 的漸進策略讓這次升級變成一個低風險的重構，而不是一次大改寫

下一篇

下一篇開始加第一個內建 middleware，Logging，用它驗證 pipeline 的 before/after 行為，也讓框架在開發時更好用

參考資料

• Ktor 官方文件：Engines
• Ktor 原始碼：SuspendFunctionGun.kt
• Ktor 原始碼：Pipeline.kt
• Ktor 原始碼：EmbeddedServer.kt
• Kotlin 官方文件：Coroutines basics
• Kotlin 官方文件：Composing suspending functions
• Kotlin 官方文件：Coroutine context and dispatchers
• Kotlin API：runBlocking
• KotlinConf 2017 - Deep Dive into Coroutines on JVM by Roman Elizarov

同步刊登於 Blog

圖片來源：AI 產生